package cn.jly.flink.operators;

import cn.jly.flink.entity.Department;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @PackageName cn.jly.flink.operators
 * @ClassName UnionDemo
 * @Description
 * @Author 姬岚洋
 * @Date 2021/1/27 下午 4:06
 */
public class UnionDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        final DataStreamSource<Department> dataStream1 = env.fromElements(
                new Department(1001, "财务部"),
                new Department(1002, "研发部")
        );

        final DataStreamSource<Department> dataStream2 = env.fromElements(
                new Department(1003, "人事部"),
                new Department(1004, "董事会")
        );

        final DataStream<Department> dataStream = dataStream1.union(dataStream2);

        dataStream
                .flatMap(new FlatMapFunction<Department, String>() {
                    @Override
                    public void flatMap(Department value, Collector<String> out) throws Exception {
                        out.collect(value.getDepartmentName());
                    }
                })
                .print();

        FlinkUtils.executeStream(env, "UnionDemo");
    }
}
